﻿using System;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using O1.Kernel.Primitives;
using O1.Kernel.Storage.Transactions;

namespace O1.Kernel.Storage
{
    internal sealed class MemoryTrunk : IDisposable
    {
        private readonly object masterSync = new object();

        private readonly StorageSequence vertexSequence;

        private readonly StorageSequence propertySequence;

        private readonly MultiVersionGCQueue vertexGcQueue;

        private readonly MultiVersionGCQueue propertyGcQueue;

        private readonly MultiVersionArray<byte[]> vertexStore;

        private readonly MultiVersionArray<byte[]> propertyStore;

        public MemoryTrunk(int initialCapacity, ushort trunkId)
        {
            this.TrunkId = trunkId;
            this.vertexSequence = new StorageSequence();
            this.propertySequence = new StorageSequence();
            this.MasterStore = new BlockArray<byte[]>(initialCapacity);
            this.vertexStore = new MultiVersionArray<byte[]>(initialCapacity);
            this.propertyStore = new MultiVersionArray<byte[]>(initialCapacity);
            this.vertexGcQueue = new MultiVersionGCQueue(this.vertexStore);
            this.propertyGcQueue = new MultiVersionGCQueue(this.propertyStore);
        }

        public ushort TrunkId { get; private set; }

        public BlockArray<byte[]> MasterStore { get; private set; }

        public long AllocateVertex(byte[] vertex, ConcurrentScope token)
        {
            if (token.Invalid)
            {
                return -1;
            }

            var index = this.vertexSequence.NextId();
            var versionId = token.TransactionId;

            // Retry loop here since allocation is "retry-able" if an attempt loses a race with another thread.
            // There is no read conflict implication to allocation. This is unlikely since the sequence is 
            // an interlocked counter.
            var owningTx = this.vertexStore.Write(index, vertex, versionId, MultiVersion.Watermark);
            while (owningTx != versionId)
            {
                index = this.vertexSequence.NextId();
                owningTx = this.vertexStore.Write(index, vertex, versionId, MultiVersion.Watermark);
            }

            token.NotifyVertexWritten(index, owningTx);
            if (index >= this.MasterStore.Capacity)
            {
                this.MasterStore.ExpandToIndex(index);
            }

            this.MasterStore[index] = new byte[6];
            return index;
        }

        public long AllocateVertex(byte[] vertex, ConcurrentScope token, byte[] propertyChain, out MasterCell master)
        {
            if (token.Invalid)
            {
                master = new MasterCell();
                return -1;
            }

            CellAddress propertyAddress;
            if (this.AllocateProperties(propertyChain, token, out propertyAddress) == -1)
            {
                master = new MasterCell();
                return -1;
            }

            var index = this.vertexSequence.NextId();
            var txId = token.TransactionId;
            var owningTx = this.vertexStore.Write(index, vertex, txId, MultiVersion.Watermark);
            while (owningTx != txId)
            {
                index = this.vertexSequence.NextId();
                owningTx = this.vertexStore.Write(index, vertex, txId, MultiVersion.Watermark);
            }

            token.NotifyVertexWritten(index, owningTx);

            if (this.MasterStore.Length <= index)
            {
                this.MasterStore.ExpandToIndex(index);
            }

            var buffer = new byte[6];
            propertyAddress.Write(buffer, 0);
            this.MasterStore[index] = buffer;
            master = new MasterCell(propertyAddress);
            return index;
        }

        public byte[] ReadVertex(long id, ConcurrentScope token)
        {
            return this.vertexStore.Read(id, token.TransactionId);
        }

        public void WriteVertex(long id, byte[] vertex, ConcurrentScope token)
        {
            if (token.Invalid)
            {
                return;
            }

            var txId = token.TransactionId;
            var owningTx = this.vertexStore.Write(id, vertex, txId, MultiVersion.Watermark);
            token.NotifyVertexWritten(id, owningTx);
        }

        public void DeleteVertex(long id, ConcurrentScope token)
        {
            var txId = token.TransactionId;
            var owningTx = this.vertexStore.Delete(id, txId, MultiVersion.Watermark);
            token.NotifyPropertiesWritten(id, owningTx);
        }

        public void WriteProperties(long id, byte[] propertyChain, ConcurrentScope token)
        {
            if (token.Invalid)
            {
                return;
            }

            var txId = token.TransactionId;
            var owningTx = this.propertyStore.Write(id, propertyChain, txId, MultiVersion.Watermark);
            token.NotifyPropertiesWritten(id, owningTx);
        }

        public long AllocateProperties(byte[] propertyChain, ConcurrentScope token, out CellAddress address)
        {
            if (token.Invalid)
            {
                address = new CellAddress();
                return -1;
            }

            var index = this.propertySequence.NextId();
            var txId = token.TransactionId;
            var owningTx = this.propertyStore.Write(index, propertyChain, txId, MultiVersion.Watermark);
            token.NotifyPropertiesWritten(index, owningTx);
            if (owningTx == txId)
            {
                address = new CellAddress((ushort)this.TrunkId, (uint)index);
                return index;
            }

            this.propertySequence.Recycle(index);
            address = new CellAddress();
            return -1;
        }

        public void DeleteProperties(long id, ConcurrentScope token)
        {
            var txId = token.TransactionId;
            var owningTx = this.propertyStore.Delete(id, txId, MultiVersion.Watermark);
            token.NotifyPropertiesWritten(id, owningTx);
        }

        public void AddShard(long vertexId, CellAddress shardAddress)
        {
            Contract.Requires(vertexId >= 0);

            // This is an infrequent operation with no concurrency control other than this coarse grained lock:
            // TODO: CAS this, check for master record reference match, and fail if concurrency is lost.
            lock (this.masterSync)
            {
                var data = this.ReadMasterCellSync(vertexId);
                var buffer = new byte[data.Length + CellAddress.Size];
                this.WriteMasterCellSync(vertexId, buffer);
            }
        }

        public byte[] ReadMasterRecord(long id)
        {
            Contract.Requires(id >= 0);

            return this.ReadMasterCellSync(id);
        }

        public byte[] ReadProperties(long propertiesId, ConcurrentScope token)
        {
            return this.propertyStore.Read(propertiesId, token.TransactionId);
        }

        public void Dispose()
        {
            this.propertyGcQueue.Dispose();
            this.vertexGcQueue.Dispose();
            GC.SuppressFinalize(this);
        }

        public void TryCommit(ConcurrentScope token)
        {
            // NOTE: Something may be written to the store but never committed and never reclaimed.
            // On load, uncommitted transactions must be scavenged as these transactions were never hardened.
            if (token.WrittenProperties != null)
            {
                if (!this.propertyStore.TryCommit(token.WrittenProperties))
                {
                    token.Invalidate();
                    this.propertyGcQueue.EnqueueForGC(token.WrittenProperties);
                    this.vertexGcQueue.EnqueueForGC(token.WrittenVertices);
                    return;
                }

                this.propertyGcQueue.EnqueueForGC(token.WrittenProperties);
            }

            if (token.WrittenVertices != null)
            {
                if (!this.vertexStore.TryCommit(token.WrittenVertices))
                {
                    token.Invalidate();
                    if (token.WrittenProperties != null)
                    {
                        this.propertyStore.TryRollback(token.WrittenProperties);
                    }
                }

                this.vertexGcQueue.EnqueueForGC(token.WrittenVertices);
            }
        }

        private byte[] ReadMasterCellSync(long vertexId)
        {
            lock (this.masterSync)
            {
                return this.MasterStore[vertexId];
            }
        }

        private void WriteMasterCellSync(long vertexId, byte[] cell)
        {
            lock (this.masterSync)
            {
                this.MasterStore[vertexId] = cell;
            }
        }
    }
}